package cn.felord.debezium.config;

import cn.felord.debezium.engine.OperationColumnHandleEngine;
//import cn.felord.debezium.engine.OperationHandleEngine;
import io.debezium.connector.mysql.MySqlConnector;
import io.debezium.data.Envelope;
import io.debezium.embedded.Connect;
import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.RecordChangeEvent;
import io.debezium.engine.format.ChangeEventFormat;
import io.debezium.engine.format.Json;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;
import java.util.Map;

import static io.debezium.data.Envelope.FieldName.*;
import static java.util.stream.Collectors.toMap;


/**
 * The type Debezium configuration.
 *
 * @author n1
 * @since 2021 /6/1 17:01
 */
@Configuration
@Slf4j
@EnableConfigurationProperties(DebeziumProperties.class)
public class DebeziumConfiguration {

//    @Resource
//    private cn.felord.debezium.engine.OperationHandleEngine OperationHandleEngine;
    @Resource
    private OperationColumnHandleEngine operationColumnHandleEngine;

    /**
     * Debezium 配置.
     *
     * @return configuration
     */
    @Bean
    io.debezium.config.Configuration debeziumConfig(DebeziumProperties properties) {
        return io.debezium.config.Configuration.create()
                .with("connector.class", MySqlConnector.class.getName())
                .with("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore")
                .with("include.schema.changes", "false")
                .with("database.history", "io.debezium.relational.history.FileDatabaseHistory")
                .with("offset.storage.file.filename", properties.getOffsetFilename())
                .with("offset.flush.interval.ms", properties.getOffsetFlush())
                .with("name", properties.getName())
                .with("database.hostname", properties.getDatabaseHost())
                .with("database.port", properties.getDatabasePort())
                .with("database.user", properties.getDatabaseUser())
                .with("database.password",properties.getDatabasePassword())
                .with("database.include.list",properties.getDatabaseIncludeList())
                .with("table.include.list", properties.getTableIncludeList())
                .with("database.server.id", properties.getDatabaseServerId())
                .with("database.server.name", properties.getDatabaseServerName())
                .with("database.history.file.filename", properties.getHistoryFilename())
                .build();
    }

    /**
     * Debezium server bootstrap debezium server bootstrap.
     *
     * @param configuration the configuration
     * @return the debezium server bootstrap
     */
    @Bean
    DebeziumServerBootstrap debeziumServerBootstrap(io.debezium.config.Configuration configuration) {
        DebeziumServerBootstrap debeziumServerBootstrap = new DebeziumServerBootstrap();
        DebeziumEngine<RecordChangeEvent<SourceRecord>> debeziumEngine = DebeziumEngine.create(ChangeEventFormat.of(Connect.class))
                .using(configuration.asProperties())
                .notifying(this::handlePayload)
                .build();

        debeziumServerBootstrap.setDebeziumEngine(debeziumEngine);
        return debeziumServerBootstrap;
    }
//    @Bean
//    DebeziumServerBootstrap debeziumServerBootstrap(io.debezium.config.Configuration configuration) {
//        DebeziumServerBootstrap debeziumServerBootstrap = new DebeziumServerBootstrap();
//        DebeziumEngine<ChangeEvent<String, String>> build = DebeziumEngine.create(Json.class)
//                .using(configuration.asProperties())
//                .notifying(record -> {
//                    System.out.println(record);
//                })
//                .build();
//
//        debeziumServerBootstrap.setDebeziumEngine(build);
//        return debeziumServerBootstrap;
//    }


    private void handlePayload(List<RecordChangeEvent<SourceRecord>> recordChangeEvents, DebeziumEngine.RecordCommitter<RecordChangeEvent<SourceRecord>> recordCommitter) {

        recordChangeEvents.forEach(r -> {
            SourceRecord sourceRecord = r.record();
            Object value = sourceRecord.value();
            Struct sourceRecordChangeValue = (Struct) sourceRecord.value();

            if (sourceRecordChangeValue != null) {
                // 判断操作的类型 过滤掉读 只处理增删改   这个其实可以在配置中设置
                Envelope.Operation operation = Envelope.Operation.forCode((String) sourceRecordChangeValue.get(OPERATION));
                Struct source = (Struct)sourceRecordChangeValue.get(SOURCE);
                String table = (String)source.get("table");
                if (operation != Envelope.Operation.READ) {
                    String record = operation == Envelope.Operation.DELETE ? BEFORE : AFTER;
                    // 获取增删改对应的结构体数据
                    Struct struct = (Struct) sourceRecordChangeValue.get(record);
                    // 将变更的行封装为Map
                    Map<String, Object> payload = getStringObjectMap(struct);
                    // 这里简单打印一下
//                    System.out.println("payload = " + payload);
//                    log.info("table:"+table+"|operate:"+operation+"|record:"+payload);
                    // 指定字段监控
                    if (operation.toString().equals("UPDATE")){
                        Struct before = (Struct) sourceRecordChangeValue.get(BEFORE);
                        Struct after = (Struct) sourceRecordChangeValue.get(AFTER);
                        operationColumnHandleEngine.handle(table,before,after);
                    }else {
                        operationColumnHandleEngine.handle(table,operation.toString(),payload);
                    }
//

                }
            }
        });
    }

    private Map<String, Object> getStringObjectMap(Struct struct) {
        return struct.schema().fields().stream()
                                .map(Field::name)
                                .filter(fieldName -> struct.get(fieldName) != null)
                                .map(fieldName -> Pair.of(fieldName, struct.get(fieldName)))
                                .collect(toMap(Pair::getKey, Pair::getValue));
    }

}
